package com.bw.hive.udtf;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.json.JSONArray;

import java.util.ArrayList;

public class ExplodeJSONArray extends GenericUDTF {


    @Override
    public void close() throws HiveException {

    }

    @Override
    public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
        if(argOIs.length != 1){
            throw new UDFArgumentException("explode_json_array 只需要一个参数");
        }
        if(argOIs[0].getCategory()!=ObjectInspector.Category.PRIMITIVE){
            throw new UDFArgumentException("explode_json_array 只能接受基础类型参数");
        }
        PrimitiveObjectInspector argumentOI = (PrimitiveObjectInspector) argOIs[0];

        if(argumentOI.getPrimitiveCategory()!= PrimitiveObjectInspector.PrimitiveCategory.STRING){
            throw new UDFArgumentException("explode_json_array 只能接受字符串参数");
        }

        ArrayList<String> fieldNames = new ArrayList<String>();
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
        fieldNames.add("items");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,
                fieldOIs);
    }

    @Override
    public void process(Object[] args) throws HiveException {
        String jsonArray = args[0].toString();
        JSONArray actions = new JSONArray(jsonArray);
        for (int i=0;i<actions.length();i++){
            String[] result = new String[1];
            result[0]=actions.getString(i);
            forward(result);
        }
    }
}
